package redis.filter;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.redis.bolt.RedisFilterBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisFilterMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import redis.WordCountBolt;
import redis.WordSpout;
import redis.search.PrintBolt;

public class FilterWordCount {
    public static final String SET_KEY_NAME = "whitelist";

    private static class WhiteListFilterMapper implements RedisFilterMapper {
        private final RedisDataTypeDescription description;

        public WhiteListFilterMapper() {
            description = new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.SET, SET_KEY_NAME);
        }
        
        @Override
        public RedisDataTypeDescription getDataTypeDescription() {
            return description;
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            // 将过滤后的word发送出去
            outputFieldsDeclarer.declare(new Fields("word"));
        }

        @Override
        public String getKeyFromTuple(ITuple iTuple) {
            return iTuple.getStringByField("word");
        }

        @Override
        public String getValueFromTuple(ITuple iTuple) {
            return null;
        }
    }

    public static void main(String[] args) throws Exception {
        JedisPoolConfig liuConfig = new JedisPoolConfig.Builder().setHost("liu").setPort(6379).build();
        WordSpout wordSpout = new WordSpout();
        WhiteListFilterMapper whiteListFilterMapper = new WhiteListFilterMapper();
        RedisFilterBolt redisFilterBolt = new RedisFilterBolt(liuConfig, whiteListFilterMapper);
        WordCountBolt wordCountBolt = new WordCountBolt();
        PrintBolt printBolt = new PrintBolt();

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("wordSpout", wordSpout);
        builder.setBolt("redisFilterBolt", redisFilterBolt).shuffleGrouping("wordSpout");
        builder.setBolt("wordCountBolt", wordCountBolt).fieldsGrouping("redisFilterBolt", new Fields("word"));
        builder.setBolt("printBolt", printBolt).shuffleGrouping("wordCountBolt");

        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("filter-topology", config, builder.createTopology());
    }
}
